package com.allenjsl.springactuator.service;
//
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.allenjsl.springactuator.entity.Credit;
//import com.allenjsl.springactuator.entity.User;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
//import org.apache.rocketmq.spring.core.RocketMQListener;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Component;
//import org.springframework.stereotype.Service;
//
//import java.util.UUID;
//
///**
// * @ClassName ConsumerService
// * @Description 消费者服务
// * @Author panchen
// * @CreateDate 2021/4/19 4:34 下午
// * @Version V1.0
// */
//@Slf4j
//@Component
//public class ConsumerService {
////    @Autowired
////    private CreditService creditService;
//    // topic需要和生产者的topic一致，consumerGroup属性是必须指定的，内容可以随意
//    // selectorExpression的意思指的就是tag，默认为“*”，不设置的话会监听所有消息
//    @Service
//    @RocketMQMessageListener(topic = "PM_TEST_TOPIC", selectorExpression = "tag1", consumerGroup = "Con_Group_One")
//    public class ConsumerSend implements RocketMQListener<User> {
//        // 监听到消息就会执行此方法
//        @Override
//        public void onMessage(User user) {
//            log.info("监听到消息：user={}", JSON.toJSONString(user));
//            Credit credit= new Credit();
//            String uuid = UUID.randomUUID().toString();
//            credit.setId(uuid);
//            credit.setU_id(user.getId());
//            credit.setU_name(user.getName());
//            credit.setIntegration(new Double(10));
//            int i = 1;
//            if (1==i){
//                log.info("# MQ消费 # 用户增加积分成功，user={}", JSONObject.toJSONString(user));
//            }else{
//                log.info("# MQ消费 # 用户增加积分失败，user={}", JSONObject.toJSONString(user));
//            }
//
//        }
//    }
//
////    // 注意：这个ConsumerSend2和上面ConsumerSend在没有添加tag做区分时，不能共存，不然生产者发送一条消息，这两个都会去消费，类型不同会有一个报错，所以实际运用中最好加上tag，写这只是让你看知道就行
////    @Service
////    @RocketMQMessageListener(topic = "PM_TEST_TOPIC", consumerGroup = "Con_Group_Two")
////    public class ConsumerSend2 implements RocketMQListener<String> {
////        @Override
////        public void onMessage(String str) {
////            log.info("监听到消息：str={}", str);
////        }
////    }
//
//    // MessageExt：是一个消息接收通配符，不管发送的是String还是对象，都可接收，当然也可以像上面明确指定类型
//    @Service
//    @RocketMQMessageListener(topic = "PM_TEST_TOPIC", selectorExpression = "tag2", consumerGroup = "Con_Group_Three")
//    public class Consumer implements RocketMQListener<MessageExt> {
//        @Override
//        public void onMessage(MessageExt messageExt) {
//            byte[] body = messageExt.getBody();
//            String msg = new String(body);
//            log.info("监听到消息：msg={}", msg);
//        }
//    }
//}


import com.allenjsl.springactuator.rocketmq.OrderMessageListener;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

@Service
public class ConsumerService {
    @Value("${server.namesrvAddr}")
    private String namesrvAddr;

    private final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("DefaultConsumer");

    @PostConstruct
    public void start() {
        try {
            //设置namesrv地址
            consumer.setNamesrvAddr(namesrvAddr);

            //从消息队列头部开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            //集群消费模式
            consumer.setMessageModel(MessageModel.CLUSTERING);

            //订阅主题
            consumer.subscribe("PM_TEST_TOPIC", "*");

            //注册消息监听器，这里因为要实现顺序消费，所以必须注册MessageListenerOrderly
            consumer.registerMessageListener(new OrderMessageListener());

            //启动消费端
            consumer.start();

            System.err.println("Order Message Consumer Start...");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
